package com.qch.kafka.demo.producer;

import com.qch.kafka.demo.entity.User;
import com.qch.kafka.demo.partitioner.MyPartitioner;
import com.qch.kafka.demo.serialization.Userserializer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Future;

/**
 * @Author qch
 * @Date 2021/5/31
 */
public class MyProducer {
    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.142.135:9092");
        configs.put("key.serializer",StringSerializer.class);
        configs.put("value.serializer",Userserializer.class);
        configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyPartitioner.class);
        //如果有多个拦截器，则设置为多个拦截器的全限定类名，中间用逗号隔开
        configs.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"com.qch.kafka.demo.interceptor.InterceptorOne,"+"com.qch.kafka.demo.interceptor.InterceptorTwo,"+"com.qch.kafka.demo.interceptor.InterceptorThree");
        configs.put("classContent","this is qch's kafka class");
        User user=new User("哈哈",1001);
        KafkaProducer<String, User> producer = new KafkaProducer<String, User>(configs);
        ProducerRecord<String, User> record = new ProducerRecord<String, User>(
                "tp_user_01",
               0,
               user.getName(),
                user
        );
       producer.send(record, new Callback() {
           public void onCompletion(RecordMetadata recordMetadata, Exception e) {
               if(e!=null){
                   System.out.println("消息发送异常");
               }else {
                   System.out.println("主题："+recordMetadata.topic()+"\t"+
                           "分区："+recordMetadata.partition()+"\t"+
                           "偏移量："+recordMetadata.offset()
                   );
               }
           }
       });
        producer.close();
    }
}
